package com.at.bigdata.spark.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 *
 * @author cdhuangchao3
 * @date 2023/5/19 10:02 PM
 */
object Spark04_Req1_HotCategoryTop10Analysis_acc {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("operator")
    val sc = new SparkContext(sparkConf)

    // Q1: rdd重复使用
    // Q2: 性能可能较低
    // Q3: 存在大量reduceByKey， 聚合算子，Spark会提供优化

    // 1、读取原始数据
    val actionRDD = sc.textFile("datas/user_visit_action.txt")

    // 2、数据结构转换
    val accumulator = new HotCategoryAccumulator
    sc.register(accumulator, "hotCategory")
    actionRDD.foreach(
      action => {
        val datas = action.split("_")
        if (datas(6) != "-1") {
          accumulator.add(datas(6), "click")
        } else if (datas(8) != "null") {
          val cids = datas(8).split(",")
          cids.foreach(cid => accumulator.add(cid, "order"))
        } else if (datas(10) != "null") {
          val cids = datas(10).split(",")
          cids.foreach(cid => accumulator.add(cid, "pay"))
        }
      }
    )
    val categories = accumulator.value.map(_._2)

    // 4、降序排列取前10
    val resultRDD = categories.toList.sortWith(
      (left, right) => {
        if (left.clickCnt > right.clickCnt) {
          true
        } else if (left.clickCnt == right.clickCnt) {
          if (left.orderCnt > right.orderCnt) {
            true
          } else if (left.orderCnt == right.orderCnt) {
            left.payCnt > right.payCnt
          } else {
            false
          }
        } else {
          false
        }
      }
    )
    // 5、将结果采集到控制台打印出来
    resultRDD.foreach(println)
    sc.stop()
  }

  case class HotCategory(cid: String, var clickCnt: Int, var orderCnt: Int, var payCnt: Int)

  class HotCategoryAccumulator extends AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] {

    var hcMap = mutable.Map[String, HotCategory]()

    override def isZero: Boolean = hcMap.isEmpty

    override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] = new HotCategoryAccumulator

    override def reset(): Unit = hcMap.clear()

    override def add(v: (String, String)): Unit = {
      val cid = v._1
      val actionType = v._2
      val category = hcMap.getOrElse(cid, HotCategory(cid, 0, 0, 0))
      if (actionType == "click") {
        category.clickCnt += 1
      } else if (actionType == "order") {
        category.orderCnt += 1
      } else if (actionType == "pay") {
        category.payCnt += 1
      }
      hcMap.update(cid, category)
    }

    override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]): Unit = {
      val map1 = this.hcMap
      val map2 = other.value
      map2.foreach{
        case (cid, hc) => {
          val category = map1.getOrElse(cid, HotCategory(cid, 0, 0, 0))
          category.clickCnt += hc.clickCnt
          category.orderCnt += hc.orderCnt
          category.payCnt += hc.payCnt
          hcMap.update(cid, category)
        }
      }
    }

    override def value: mutable.Map[String, HotCategory] = hcMap
  }
}
